
#pragma once

#include "mosquitto.h"
#include "spdlog/spdlog.h"
#include <atomic>
#include <condition_variable>
#include <functional>
#include <iostream>
#include <mutex>
#include <queue>
#include <string>
#include <vector>

class MqttClient {
public:
    class Msg_t {
    public:
        std::string topic;
        std::string payload;

    public:
        Msg_t() = delete;
        Msg_t(Msg_t &&) = default;

        Msg_t(const std::string &t, const std::string &p)
            : topic(t)
            , payload(p)
        {
        }

        Msg_t(const Msg_t &) = default;

        Msg_t &operator=(const Msg_t &) = default;

        ~Msg_t() = default;
    };

    std::vector<std::function<bool()>> doConnected; /* 在连接的时候要做的事情 */
    std::string _clientId;                          /* 客户端标识 */
private:
    std::atomic<bool> _isConnected; /* 是否连接标志 */
    struct mosquitto *_mosquitto;   /* 句柄 */
public:
    /* 消息队列 */
    std::queue<Msg_t> _mqttMsgQueue;
    std::mutex _mutex;
    std::condition_variable _cv;

private:
    static void onConnect(struct mosquitto *mosq, void *userData, int result);
    static void onDisconnect(struct mosquitto *mosq, void *userData, int result);
    static void onMessage(struct mosquitto *mosq, void *userData, const mosquitto_message *message);

public:
    MqttClient(const std::string &clientId);

    MqttClient() = delete;                   /* 不允许无参构造 */
    MqttClient(const MqttClient &) = delete; /* 不允许拷贝构造 */
    MqttClient &operator=(const MqttClient &) = delete;
    MqttClient(MqttClient &&) = delete; /* 不允许移动拷贝 */

    bool connect(const std::string &host, int port = 1883, int keepAlive = 60);
    bool connect(const std::string &host, const std::string &userName, const std::string &passwd, int port = 1883, int keepAlive = 60);

    bool disConnect(void);

    bool publish(const std::string &topic, const std::string &payload);

    bool subscribe(const std::string &topic);

    bool unSubscribe(const std::string &topic);

    bool isConnected(void);

    void registerOnConnected(const std::function<bool()> &func);

    Msg_t getOneItem(void);

    ~MqttClient();
};

MqttClient::MqttClient(const std::string &clientId)
    : _clientId(clientId)
    , _isConnected(false)
    , _mosquitto(nullptr)
{
    mosquitto_lib_init();

    _mosquitto = mosquitto_new(_clientId.c_str(), true, this);
    if (!_mosquitto) {
        throw std::runtime_error("Failed to create Mosquitto instance");
    }

    mosquitto_connect_callback_set(_mosquitto, onConnect);
    mosquitto_disconnect_callback_set(_mosquitto, onDisconnect);
    mosquitto_message_callback_set(_mosquitto, onMessage);
}

bool MqttClient::connect(const std::string &host, int port, int keepAlive)
{
    if (_isConnected) {
        return true;
    }

    mosquitto_connect(_mosquitto, host.c_str(), port, keepAlive);

    return !mosquitto_loop_start(_mosquitto);
}

bool MqttClient::connect(const std::string &host, const std::string &userName, const std::string &passwd, int port, int keepAlive)
{
    if (_isConnected) {
        return true;
    }

    mosquitto_username_pw_set(_mosquitto, userName.c_str(), passwd.c_str());

    mosquitto_connect(_mosquitto, host.c_str(), port, keepAlive);

    return !mosquitto_loop_start(_mosquitto);
}

bool MqttClient::disConnect(void)
{
    if (!_isConnected) {
        return true;
    }

    mosquitto_disconnect(_mosquitto);
    return !mosquitto_loop_stop(_mosquitto, true);
}

bool MqttClient::publish(const std::string &topic, const std::string &payload)
{
    // std::cout << "publish topic: " << topic << std::endl;
    // std::cout << "publish payload: " << payload << std::endl;
    spdlog::info("publish topic: {}", topic);
    spdlog::info("publish payload: {}", payload);

    return !mosquitto_publish(_mosquitto, nullptr, topic.c_str(), payload.length(), payload.c_str(), 0, false);
}

bool MqttClient::subscribe(const std::string &topic)
{
    // std::cout << "subscribe topic: " << topic << std::endl;
    spdlog::info("subscribe topic: {}", topic);

    return !mosquitto_subscribe(_mosquitto, nullptr, topic.c_str(), 0);
}

bool MqttClient::unSubscribe(const std::string &topic)
{
    // std::cout << "unSubscribe topic: " << topic << std::endl;
    spdlog::info("unSubscribe topic: {}", topic);
    return !mosquitto_unsubscribe(_mosquitto, nullptr, topic.c_str());
}

void MqttClient::onConnect(struct mosquitto *mosq, void *userData, int result)
{
    if (result == MOSQ_ERR_SUCCESS) {
        MqttClient *client = reinterpret_cast<MqttClient *>(userData);
        client->_isConnected = true;

        // std::cout << client->_clientId << " connect success" << std::endl;
        spdlog::info("{} connect success", client->_clientId);

        for (auto &func : client->doConnected) {
            func();
        }

    } else {
        // std::cerr << "Failed to connect to MQTT broker: " << mosquitto_connack_string(result) << std::endl;
        spdlog::error("Failed to connect to MQTT broker: {}", mosquitto_connack_string(result));
    }
}

void MqttClient::onDisconnect(struct mosquitto *mosq, void *userData, int result)
{
    MqttClient *client = reinterpret_cast<MqttClient *>(userData);
    client->_isConnected = false;
}

void MqttClient::onMessage(struct mosquitto *mosq, void *userData, const mosquitto_message *message)
{
    MqttClient *client = reinterpret_cast<MqttClient *>(userData);
    // std::cout << "recv topic: " << static_cast<char *>(message->topic) << std::endl;
    // std::cout << "recv payload: " << static_cast<char *>(message->payload) << std::endl;

    spdlog::info("{} recv topic: {}", client->_clientId, static_cast<char *>(message->topic));
    spdlog::info("{} recv payload: {}", client->_clientId, static_cast<char *>(message->payload));

    {
        std::unique_lock<std::mutex> _lock(client->_mutex);

        client->_mqttMsgQueue.emplace(Msg_t(static_cast<char *>(message->topic), static_cast<char *>(message->payload)));

        client->_cv.notify_one();
    }
}

MqttClient::Msg_t MqttClient::getOneItem(void)
{
    std::unique_lock<std::mutex> _lock(_mutex);

    _cv.wait(_lock, [this]() { return !this->_mqttMsgQueue.empty(); });

    Msg_t msg = _mqttMsgQueue.front();

    _mqttMsgQueue.pop();

    return msg;
}

void MqttClient::registerOnConnected(const std::function<bool()> &func)
{
    if (func) {
        doConnected.emplace_back(func);
    }
}

bool MqttClient::isConnected(void)
{
    return _isConnected;
}

MqttClient::~MqttClient()
{
    if (_mosquitto) {
        mosquitto_destroy(_mosquitto);
        mosquitto_lib_cleanup();
    }
}